package com.baishancloud.log.traffic.sjz

import com.baishancloud.log.common.util.other.DataTimeFilterUtil
import com.baishancloud.log.format.scala.Origin
import com.baishancloud.log.traffic.{filterDomains, filterSecond}
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration

import java.util.Objects

/**
 * 对指定的域名进行最小时间过滤，该过滤适用于石家庄地区运行的任务
 *
 * @author ziqiang.wang 
 * @date 2022/3/31 17:56
 */
class DataFilterSjz(parameterTool: ParameterTool) extends RichFilterFunction[Origin] {

  var filterDomainsSet: Set[String] = _
  var minSecond: Long = Long.MinValue

  override def open(parameters: Configuration): Unit = {
    filterDomainsSet = parameterTool.get(filterDomains).split(",").toSet
    minSecond = parameterTool.getLong(filterSecond)
  }


  /**
   * 在原先的基础上，增加对指定域名的时间过滤，该类对应任务运行在石家庄地区服务器，指定时间为消费的最小时间，判断条件为 >=
   */
  override def filter(value: Origin): Boolean = {
    Objects.equals(value.xPeer(), "-") &&
      value.url() != null &&
      value.url().contains("bsxdisp=se") &&
      DataTimeFilterUtil.domainSecondFilterMin(value, filterDomainsSet, minSecond)
  }
}
